/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.rainsoft.center.isec.common.utils;

import com.rainsoft.center.isec.common.entity.Constants;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class Producer extends Thread {
	//生产消息的条数
	public static final Integer nums = 11;
	//自定义topicName;
	public static final String topicName = "realid_loginout";
	private final KafkaProducer<Integer, String> producer;
	private final String topic;
	private final Boolean isAsync;

	public Producer(String topic, Boolean isAsync) {
		Properties props = new Properties();
		props.put("bootstrap.servers", Constants.KAFKA_BROKERS);
		props.put("client.id", "DemoProducer");
		props.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		producer = new KafkaProducer<>(props);
		this.topic = topic;
		this.isAsync = isAsync;
	}

	/*" + Math.round(Math.random() * 10) + "*/
	public void run() {
		int messageNo = 1;
		while (true) {
			//控制生产条数用
			if (messageNo == Producer.nums) {
				System.exit(0);
			}
			//	Utils.sleep(1000);
			String messageStr ="{\"ICCARD_TYPE\":\"11\",\"LOGIN_TIME\":\""+new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())+"\"," +
					"\"CHECKIN_ID\":\"11022399999906411115210419880630091020171109094400\",\"IDTYPE\":\"11"+Math.round(Math.random() * 10)  +"\",\"ORG_NAME\":\"牙克石市公安局\"," +
					"\"IDCODE\":\"1521041988063009"+Math.round(Math.random() * 10)  +"\",\"SEX\":\"1\",\"SERVICE_CODE\":\"11011210900025\",\"SERVICE_CODE_OUT\":\"11022310000064\"," +
					"\"DATA_SOURCE\":\"1\",\"BLOOD_TYPE\":\"\",\"USER_ADDRESS\":\"内蒙古牙克石市迎宾西街供电局家属楼3号楼1单元3号\",\"USER_PHONE\":\"\"," +
					"\"ID_EXPIRETIME\":\"2011.03.01-2021.03.01\",\"LOGOUT_TIME\":\"\",\"ROOM_ID\":\"\",\"COUNTRY\":\"CHN\",\"FINGERPRINT\":\"\"," +
					"\"ICCARD_NO\":\"104198806300910\",\"MACHINE_CODE\":\"5964630212851320fad4c\",\"USERNAME\":\"潘帮(扫)\",\"PEOPLE\":\"HA\",\"REMARK\":\"\"," +
					"\"BIRTHDAY\":\"1988-06-30 00:00:00\"}";

			long startTime = System.currentTimeMillis();
			if (isAsync) { // Send asynchronously
				producer.send(new ProducerRecord<>(topic,
						messageNo,
						messageStr), new DemoCallBack(startTime, messageNo, messageStr));
			} else { // Send synchronously
				try {
					producer.send(new ProducerRecord<>(topic,
							messageNo,
							messageStr)).get();
					System.out.println("Sent message: (" + messageNo + ", " + messageStr + ")");
				} catch (InterruptedException | ExecutionException e) {
					e.printStackTrace();
				}
			}
			++messageNo;

		}
	}

	public static void main(String[] args) {
		Producer producerThread = new Producer(topicName, false);
		producerThread.start();
	}

}

class DemoCallBack implements Callback {

	private final long startTime;
	private final int key;
	private final String message;

	public DemoCallBack(long startTime, int key, String message) {
		this.startTime = startTime;
		this.key = key;
		this.message = message;
	}

	/**
	 * A callback method the user can implement to provide asynchronous handling of request completion. This method will
	 * be called when the record sent to the server has been acknowledged. Exactly one of the arguments will be
	 * non-null.
	 *
	 * @param metadata  The metadata for the record that was sent (i.e. the partition and offset). Null if an error
	 *                  occurred.
	 * @param exception The exception thrown during processing of this record. Null if no error occurred.
	 */
	public void onCompletion(RecordMetadata metadata, Exception exception) {
		long elapsedTime = System.currentTimeMillis() - startTime;
		if (metadata != null) {
			System.out.println(
					"message(" + key + ", " + message + ") sent to partition(" + metadata.partition() +
							"), " +
							"offset(" + metadata.offset() + ") in " + elapsedTime + " ms");

		} else {
			exception.printStackTrace();
		}
	}
}
